package com.yupi.springbootinit.bimq;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.rabbitmq.client.Channel;
import com.yupi.springbootinit.common.ErrorCode;
import com.yupi.springbootinit.constant.CommonConstant;
import com.yupi.springbootinit.constant.MqConstant;
import com.yupi.springbootinit.exception.BusinessException;
import com.yupi.springbootinit.manager.AiManager;
import com.yupi.springbootinit.model.entity.Chart;
import com.yupi.springbootinit.model.entity.User;
import com.yupi.springbootinit.service.ChartService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class BiConsumer {
    @Resource
    private ChartService chartService;

    @Resource
    private AiManager aiManager;

    @Resource
    RedisTemplate redisTemplate;

    @Resource
    BiProducer biProducer;

    @SneakyThrows
    @RabbitListener(queues = {MqConstant.QUEUE_NAME}, ackMode = "MANUAL")
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTage){
        log.info("已接受到消息 tag ：" + deliveryTage + "消息内容是" +  message);
        if(StringUtils.isBlank(message)){
            channel.basicNack(deliveryTage, false, false);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "消息为空");
        }

        long chartId = Long.parseLong(message);
        Chart chart = chartService.getById(chartId);

        if(chart == null){
            channel.basicNack(deliveryTage, false, false);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "图表为空");
        }

        // 更改状态为执行中
        Chart updateChartStatus = new Chart();
        updateChartStatus.setId(chartId);
        updateChartStatus.setStatus("running");
        boolean updateStatus = chartService.updateById(updateChartStatus);

        if(!updateStatus){
            channel.basicNack(deliveryTage, false, false);
            chartService.handleChartUpdateError(chartId, "消费者消费消息时，更改状态为执行中失败");
            return;
        }

        // 有更改数据库操作，保证双写一致性发送消息
        log.info("发送消息，保证双写一致性，userid：" + chart.getUserId());
        biProducer.sendRedisMySQLConsistMessage(chart.getUserId() + "");

        // 获取 Bi 结果并校验
        String result = null;

        try {
            result = aiManager.doChat(chartService.buildUserInput(chart), CommonConstant.MODEL_ID);
        } catch (Exception e) {
            channel.basicNack(deliveryTage, false, true);
            throw  new BusinessException(ErrorCode.SYSTEM_ERROR, "ai 生成失败");
        }

        try {
            boolean saveResult = chartService.saveBIResult(result, chartId);
            if(!saveResult){
                chartService.handleChartUpdateError(chartId, "保存 chart 失败");
            }
        } catch (Exception e) {
            channel.basicNack(deliveryTage, false, true);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "保存 chart 失败");
        }

        // 有更改数据库操作，保证双写一致性发送消息
        log.info("发送消息，保证双写一致性，userid：" + chart.getUserId());
        biProducer.sendRedisMySQLConsistMessage(chart.getUserId() + "");

        log.info("已消费消息 tag ：" + deliveryTage + ", 的消息，消息内容为" + chartId);
        channel.basicAck(deliveryTage, false);
    }

    static final String PAGE_CACHE = "my_chart_page";

    static final Long PAGE_CACHE_TIME = new Long(5);

    @SneakyThrows
    @RabbitListener(queues = {MqConstant.REDIS_MYSQL_CONSIST_QUEUE_NAME}, ackMode = "MANUAL")
    public void receiveRedisMySQLConsistMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTage){
        log.info("保证 redis 和 mysql 双写一致性，开始消费 tag：" + deliveryTage + "的信息，消息的内容是：" + message);

        if(StringUtils.isBlank(message)){
            log.info("消费 tag：" + deliveryTage + "的信息失败，message 为空");
            channel.basicNack(deliveryTage, false, false);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "message 为空");
        }

        // 删除缓存
        Long pageSize = new Long(4);
        Long pageNum = new Long(1);

        String userKey = PAGE_CACHE + ":" + message;
        String pageKey =  PAGE_CACHE + ":" + pageNum + ":" + pageSize;
        redisTemplate.delete(userKey);

        // 获取最新信息
        String sortField = "createTime";
        String sortOrder = CommonConstant.SORT_ORDER_ASC;
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.eq("userId", new Long(message));
        queryWrapper.orderBy(true, sortOrder == CommonConstant.SORT_ORDER_DESC, sortField);
        Page<Chart> dbResult = chartService.page(new Page<>(pageNum, pageSize), queryWrapper);

        // 更新信息
        redisTemplate.opsForHash().put(userKey, pageKey, dbResult);
        redisTemplate.expire(userKey, PAGE_CACHE_TIME, TimeUnit.MINUTES);

        log.info("保证 redis 和 mysql 双写一致性，已经消费完成 tag：" + deliveryTage + "的信息");
        channel.basicAck(deliveryTage, false);
    }
}
